package cn.llyong.rpc.common.discovery;

import cn.llyong.rpc.common.config.MonkeyRpcProperties;
import cn.llyong.rpc.common.constant.Constants;
import cn.llyong.rpc.common.exception.RpcException;
import cn.llyong.rpc.common.utils.RpcCacheHolder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @description:
 * @author: lvyong
 * @date: 2020年06月28日 15:56
 */
@Component
public class ServiceDiscovery {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private MonkeyRpcProperties rpcProperties;

    public static CuratorFramework zkClient = null;

    private static ReentrantLock lock = new ReentrantLock();

    @PostConstruct
    public void init(){
        connect(rpcProperties.getRegistryAddr());
    }

    private  boolean connect(String registryAddr){
        zkClient = CuratorFrameworkFactory.builder()
                .connectString(registryAddr)
                .retryPolicy(new RetryNTimes(3, 5000))
//                .namespace("")
                .build();
        zkClient.start();
        return zkClient != null;
    }

    /**
     * 服务发现
     * @param serverName
     * @return
     */
    public List<String> discovery(String serverName) throws RpcException {
        isConnected();

        try {
            String serverPath = Constants.ROOT_PATH + rpcProperties.getGroupName() + "/" + serverName;
            //服务列表
            List<String> serverList = RpcCacheHolder.SUBSCRIBE_SERVICES.get(serverName);
            if (!CollectionUtils.isEmpty(serverList)) {
                return serverList;
            }
            serverList = zkClient.getChildren().forPath(serverPath);

            //同时监控节点变化
            PathChildrenCache childrenCache = new PathChildrenCache(zkClient, serverPath,true);
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
            addListener(childrenCache,serverName);

            return serverList;
        } catch (Exception e) {
            throw new RpcException(e);
        }
    }

    private void addListener(PathChildrenCache childrenCache, String serverName){
        //使用自定义的线程池
        ExecutorService threadPool = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(32), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

        Listenable<PathChildrenCacheListener> listenable = childrenCache.getListenable();
        listenable.addListener((curatorFramework, event) -> {

            // 创建子节点
            if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){
                String path = event.getData().getPath();
                String host = path.substring(path.lastIndexOf("/") + 1, path.length());
                logger.info("服务 {} 的 {} 服务器节点上线！！！", serverName, host);

                lock.lock();
                try {
                    List<String> list = RpcCacheHolder.SUBSCRIBE_SERVICES.getOrDefault(serverName, new ArrayList<>());
                    list.add(host);
                } finally {
                    lock.unlock();
                }
            }
            // 删除子节点
            else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                String path = event.getData().getPath();
                String host = path.substring(path.lastIndexOf("/") + 1, path.length());
                logger.info("服务 {} 的 {} 服务器节点上线！！！", serverName, host);

                lock.lock();
                try {
                    List<String> list = RpcCacheHolder.SUBSCRIBE_SERVICES.get(serverName);
                    list.remove(host);
                } finally {
                    lock.unlock();
                }
            }
        }, threadPool);
    }


    /**
     * 判断是否建立连接
     */
    private void isConnected(){
        if (zkClient == null){
            throw new RpcException("have not connect Zookeeper Server");
        }
    }

}
